package com.exsun.gateway.jt809.kafka;

import com.exsun.gateway.jt809.common.LinkMgr;
import com.exsun.gateway.jt809.common.LinkType;
import com.exsun.gateway.jt809.data.PushData;
import com.exsun.gateway.jt809.handler.SendMgrHandler;
import exsun.jt808.data.config.CommonMsgTopic;
import exsun.jt808.data.config.jt809.Jt809MsgIdConfig;
import exsun.jt808.data.jt809_data.SendBaseEntity;
import exsun.jt808.data.jt809_data.alarm.main.UpWarnMsgOperationInfoEx;
import exsun.jt808.data.jt809_data.vehicledynamic.main.UpExgMsgSafetyTerminal;
import exsun.jt808.data.util.GsonUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;

import java.util.ArrayList;
import java.util.Date;
import java.util.List;

public class JT809KafkaUpExgMsgSafetyTerminalHandler extends JT809KafkaHandleBase{


    private Logger logger = LoggerFactory.getLogger(JT809KafkaUpExgMsgSafetyTerminalHandler.class);
    @Autowired
    private SendMgrHandler sendMgrHandler;

    //从Kafka监听809消息
    @KafkaListener(topics = CommonMsgTopic.distributeToJT809, containerFactory = "PushDataToPlatform809GroupFactory")
    public void batchUpExgMsgSafetyTerminalListen(List<ConsumerRecord<?, ?>> records, Acknowledgment ack) {
        try {
            List<PushData> temp = done(records);
            LinkMgr.push2FitChannel(temp, LinkType.MASTER);
            //发送错误，则不提交偏移量
            if(temp == null)return;
            ack.acknowledge();
        }catch(Exception e){
            logger.info(e.getMessage());
        }
    }

    public List<PushData> done(List<ConsumerRecord<?, ?>> records) {
        List<PushData> temp = new ArrayList<>();
        for (ConsumerRecord<?, ?> record : records) {
//            //任意链路都不通，则不处理数据
//            if (!LinkMgr.getLinkStatus()) return null;
            String msg = record.value().toString();
            SendBaseEntity object = GsonUtils.getObjectFromJson(msg, SendBaseEntity.class);
            switch (object.msgId){
                case Jt809MsgIdConfig.UP_EXG_MSG_SAFETY_TERMINAL:
                    UpExgMsgSafetyTerminal upExgMsgSafetyTerminal = new UpExgMsgSafetyTerminal();
                    upExgMsgSafetyTerminal = GsonUtils.getObjectFromJson(msg,UpExgMsgSafetyTerminal.class);
                    upExgMsgSafetyTerminal.distributeTime = new Date();
                    byte[] body = sendMgrHandler.jt809Encode.UpExgMsgSafetyTerminal(upExgMsgSafetyTerminal,object.msgId);
                    PushData pushData = new PushData(Jt809MsgIdConfig.UP_EXG_MSG_SAFETY_TERMINAL, body);
                    temp.add(pushData);
                    break;
            }
        }
        return temp;
    }
}
